package cn.com.rocketmq.filter;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;

import java.util.List;

public class PushConsumer {

    /**
     * 日志对象
     */
    private static  final Logger logger = LoggerFactory.getLogger(PushConsumer.class);


    public static void main(String[] args) {
        DefaultMQPushConsumer consumer = null;
        try {
            consumer = new DefaultMQPushConsumer("group_rocketmq_syncDemo");
            consumer.setNamesrvAddr("192.168.113.101:9876");
            //参数一：topic、参数二：过滤（*表示不过滤）
            //过滤方式一
            //consumer.subscribe("rocketmq_syncDemo", "tagA || tagC");

            //过滤方式二（注意该sql里面字段是区分大小写的）
            //sql过滤方式，borker配置文件必须设置属性enablePropertyFilter=true，并且消费者必须是推模式
            //另外消息过滤行为是在broker端进行的，可以提升网络传输性能，但是会增加服务器的压力（将过滤sql推送给broker）
            consumer.subscribe("rocketmq_syncDemo", MessageSelector.bySql("TAGS is not null and TAGS in ('tagA','tagC') and type = 'elg_0'"));

            //设置消息监听
            //MessageListenerConcurrently  并发消费
            //MessageListenerOrderly 顺序消费
            consumer.setMessageListener(new MessageListenerConcurrently() {
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    list.forEach(item -> {
                        try {
                            logger.info("消息消费成功！消息ID={}，type={}，消息内容：{}", item.getMsgId(), item.getUserProperty("type"),new String(item.getBody(), "UTF-8"));
                        } catch (Exception e) {
                            logger.error("消息消费失败！", e);
                        }
                    });
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            //消费者启动
            consumer.start();
        } catch (MQClientException e) {
            logger.error("消费者消费异常！",e);
        }
    }
}
